﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using Microsoft.VisualStudio.TestPlatform.UnitTestFramework;
using XcoAppSpaces.Contracts.Logging;
using XcoAppSpaces.Contracts.Transport;
//using XcoAppSpaces.Serialization.Binary;
using XcoAppSpaces.Transport.Sockets;
using XcoAppSpaces.Contracts.Exceptions;
using System.Threading.Tasks;

namespace test.XcoAppSpaces.Transport.Sockets
{
	[TestClass]
	public class TransferTests
	{
		[TestMethod]
		public void TestTransfer_simple()
		{
			//XcoLoggerFactory.CreateInstance = (t => new TestLogger.TestLogger());
			string addr1 = "localhost" + ":8000";
			string addr2 = "localhost" + ":8001";

			var receiver1 = new ReceivedTmp();
			var receiver2 = new ReceivedTmp();
			string msg = "test";

			using (XcoTCPTransportService comm1 = new XcoTCPTransportService(8000),
				comm2 = new XcoTCPTransportService(8001))
			{
				comm1.OnMessageReceived += receiver1.MessageReceived;
				comm2.OnMessageReceived += receiver2.MessageReceived;

				comm1.Start();
				comm2.Start();

				comm1.Send(new XcoMessage {Content = Encoding.UTF8.GetBytes(msg), CausalityContext = null}, addr2);
				comm2.Send(new XcoMessage {Content = Encoding.UTF8.GetBytes(msg), CausalityContext = null}, addr1);
				comm1.Send(new XcoMessage {Content = Encoding.UTF8.GetBytes(msg), CausalityContext = null}, addr2);
				
				Task.Delay(500).Wait();

				comm1.Stop();
				comm2.Stop();
				
				Assert.AreEqual(1, receiver1.Counter);
				Assert.AreEqual(2, receiver2.Counter);
			}
		}

		[TestMethod]
		public void TestTransfer_simple_with_authentication()
		{
			XcoLoggerFactory.CreateInstance = (t => new DebugLogger());
			string addr1 = "localhost" + ":8000";
			string addr2 = "localhost" + ":8001";

			var receiver1 = new ReceivedTmp();
			var receiver2 = new ReceivedTmp();
			string msg = "test";

			using (XcoTCPTransportService comm1 = new XcoTCPTransportService(8000),
				comm2 = new XcoTCPTransportService(8001))
			{
				comm1.OnMessageReceived += receiver1.MessageReceived;
				comm2.OnMessageReceived += receiver2.MessageReceived;

				comm1.SecuritySettings.AuthenticationKey = "test";
				comm2.SecuritySettings.AuthenticationKey = "test";

				comm1.Start();
				comm2.Start();

				comm1.Send(new XcoMessage { Content = Encoding.UTF8.GetBytes(msg), CausalityContext = null }, addr2);
				comm2.Send(new XcoMessage { Content = Encoding.UTF8.GetBytes(msg), CausalityContext = null }, addr1);
				comm1.Send(new XcoMessage { Content = Encoding.UTF8.GetBytes(msg), CausalityContext = null }, addr2);

				Task.Delay(500).Wait();

				comm1.Stop();
				comm2.Stop();

				Assert.AreEqual(1, receiver1.Counter);
				Assert.AreEqual(2, receiver2.Counter);
			}
		}


		[TestMethod]
		public void TestTransfer()
		{
			//XcoLoggerFactory.CreateInstance = (t => new TestLogger.TestLogger());
			string addr1 = "localhost" + ":8000";
			string addr2 = "localhost" + ":8001";

			var receiver1 = new ReceivedTmp();
			var receiver2 = new ReceivedTmp();
			string msg = "test";

			using (XcoTCPTransportService comm1 = new XcoTCPTransportService(8000),
			                              comm2 = new XcoTCPTransportService(8001))
			{
				comm1.OnMessageReceived += receiver1.MessageReceived;
				comm2.OnMessageReceived += receiver2.MessageReceived;

				comm1.Start();
				comm2.Start();

				comm1.Send(new XcoMessage {Content = Encoding.UTF8.GetBytes(msg), CausalityContext = null}, addr2);
				comm2.Send(new XcoMessage { Content = Encoding.UTF8.GetBytes(msg), CausalityContext = null }, addr1);
				comm1.Send(new XcoMessage { Content = Encoding.UTF8.GetBytes(msg), CausalityContext = null }, addr2);
				Task.Delay(100).Wait();
				comm2.Stop();
				Task.Delay(500).Wait();

				try
				{
					var xcoMsg = new XcoMessage { Content = Encoding.UTF8.GetBytes(msg), CausalityContext = null };
					Exception expectedEx = null;
					xcoMsg.OnTransferFailed += (_, ex) => expectedEx = ex;
					comm1.Send(xcoMsg, addr2);
					Assert.IsNotNull(expectedEx, "Should not be able to send message!");
				}
				catch (XcoCommunicationException)
				{
				}
				Task.Delay(1000).Wait();
				//Assert.AreEqual(1, receiver1.Counter);
				Assert.AreEqual(2, receiver2.Counter);
			}

			using (XcoTCPTransportService comm1 = new XcoTCPTransportService(8000),
			                              comm2 = new XcoTCPTransportService(8001))
			{
				comm1.OnMessageReceived += receiver1.MessageReceived;
				comm2.OnMessageReceived += receiver2.MessageReceived;
				comm1.Start();
				comm2.Start();

				comm1.Send(new XcoMessage { Content = Encoding.UTF8.GetBytes(msg), CausalityContext = null }, addr2);
				comm2.Send(new XcoMessage { Content = Encoding.UTF8.GetBytes(msg), CausalityContext = null }, addr1);
				comm1.Send(new XcoMessage { Content = Encoding.UTF8.GetBytes(msg), CausalityContext = null }, addr2);
				Task.Delay(100).Wait();
				comm2.Stop();
				Task.Delay(500).Wait();

				try
				{
					var xcoMsg = new XcoMessage {Content = Encoding.UTF8.GetBytes(msg), CausalityContext = null};
					Exception expectedEx = null;
					xcoMsg.OnTransferFailed += (_, ex) => expectedEx = ex;
					comm1.Send(xcoMsg, addr2);
					Assert.IsNotNull(expectedEx, "Should not be able to send message!");
				}
				catch (XcoCommunicationException)
				{
				}
				Task.Delay(1000).Wait();
				//Assert.AreEqual(2, receiver1.Counter);
				Assert.AreEqual(4, receiver2.Counter);
			}
		}

		[TestMethod]
		public void TestActiveConnections()
		{
			var receiver1 = new ReceivedTmp();
			var receiver2 = new ReceivedTmp();
			var msg = "test";

			using (XcoTCPTransportService comm1 = new XcoTCPTransportService(0),
			                              comm2 = new XcoTCPTransportService(0))
			{
				Assert.IsNull(comm1.ActiveConnections);
				Assert.IsNull(comm2.ActiveConnections);

				comm1.OnMessageReceived += receiver1.MessageReceived;
				comm2.OnMessageReceived += receiver2.MessageReceived;
				comm1.Start();
				comm2.Start();
				var addr1 = comm1.LocalAddress;
				var addr2 = comm2.LocalAddress;

				Assert.AreEqual(0, comm1.ActiveConnections.Count());
				Assert.AreEqual(0, comm2.ActiveConnections.Count());

				comm1.Send(new XcoMessage {Content = Encoding.UTF8.GetBytes(msg), CausalityContext = null}, addr2);
				Task.Delay(200).Wait();
				Assert.AreEqual(1, comm1.ActiveConnections.Count());
				Assert.AreEqual(1, comm2.ActiveConnections.Count());

				comm2.Send(new XcoMessage {Content = Encoding.UTF8.GetBytes(msg), CausalityContext = null}, addr1);
				Task.Delay(100).Wait();
				Assert.AreEqual(1, comm1.ActiveConnections.Count());
				Assert.AreEqual(1, comm2.ActiveConnections.Count());


				Assert.AreEqual(addr2, comm1.ActiveConnections.First());
				Assert.AreEqual(addr1, comm2.ActiveConnections.First());
			}
		}

		[TestMethod]
		public void ConcurrentTransfer()
		{
			string addr2 = "localhost" + ":8001";

			using (XcoTCPTransportService comm1 = new XcoTCPTransportService(8000),
			                              comm2 = new XcoTCPTransportService(8001))
			{
				int received = 0;
				comm2.OnMessageReceived += (message, address, service) => Interlocked.Increment(ref received);

				comm1.Start();
				comm2.Start();

				const int sent = 20;
				int successFullyTransferred = 0;
				int transferErrors = 0;
				for (int i = 0; i < sent; i++)
				{
					System.Threading.Tasks.Task.Factory.StartNew(() =>
						{
							var msg = new XcoMessage {Content = new byte[] {1, 2, 3}, CausalityContext = null};
							msg.OnTransferSuccessful += _ => Interlocked.Increment(ref successFullyTransferred);
							msg.OnTransferFailed += (m,ex) => Interlocked.Increment(ref transferErrors);
							comm1.Send(msg, addr2);
						});
				}

				Assert.IsTrue(SpinWait.SpinUntil(() => received == sent, 1000));
				Assert.AreEqual(sent, successFullyTransferred);
				Assert.AreEqual(0, transferErrors);
			}
		}

		[TestMethod]
		public void ConcurrentFailedTransfer()
		{
			string addr2 = "localhost" + ":8001";

			using (var comm1 = new XcoTCPTransportService(8000))
			{
				comm1.Start();

				const int sent = 20;
				int successFullyTransferred = 0;
				int transferErrors = 0;
				for (int i = 0; i < sent; i++)
				{
					Task.Factory.StartNew(() =>
					{
						var msg = new XcoMessage { Content = new byte[] { 1, 2, 3 }, CausalityContext = null };
						msg.OnTransferSuccessful += _ => Interlocked.Increment(ref successFullyTransferred);
						msg.OnTransferFailed += (m, ex) => Interlocked.Increment(ref transferErrors);
						comm1.Send(msg, addr2);
					});
				}

				SpinWait.SpinUntil(() => transferErrors == sent, 5000);
				Assert.AreEqual(sent, transferErrors);
				Assert.AreEqual(0, successFullyTransferred);
			}
		}

		[TestMethod]
		public void Failed_Transfer_after_Successful_Transfer()
		{
			string addr2 = "localhost" + ":8001";

			using (XcoTCPTransportService comm1 = new XcoTCPTransportService(8000),
										  comm2 = new XcoTCPTransportService(8001))
			{
				int received = 0;
				comm2.OnMessageReceived += (message, address, service) => Interlocked.Increment(ref received);

				comm1.Start();
				comm2.Start();

				int successFullyTransferred = 0;
				int transferErrors = 0;
				var msg = new XcoMessage { Content = new byte[] { 1, 2, 3 }, CausalityContext = null };
				msg.OnTransferSuccessful += _ => Interlocked.Increment(ref successFullyTransferred);
				msg.OnTransferFailed += (m, ex) => { Interlocked.Increment(ref transferErrors); System.Diagnostics.Debug.WriteLine(ex.Message); };

				//successful send
				comm1.Send(msg, addr2);
				Task.Delay(200).Wait();

				comm2.Dispose();
				Task.Delay(300).Wait();
				
				//failed send
				comm1.Send(msg, addr2);
				Task.Delay(200).Wait();
				
				Assert.AreEqual(1, successFullyTransferred);
				Assert.AreEqual(1, transferErrors);
			}
		}
	}
}
